package com.bw.yk02.fnc;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

public class HbaseSinkFunction  extends RichSinkFunction<String> {

    private Connection con;

    public Connection getConnection() {
        try {
            Class.forName("org.apache.phoenix.jdbc.PhoenixDriver");
            return DriverManager.getConnection("jdbc:phoenix:localhost:2181");
        }catch (Exception e)
        {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        con = getConnection();
    }


    @Override
    public void invoke(String value, Context context) throws Exception {
        super.invoke(value, context);
        JSONObject jsonObject = (JSONObject) JSONObject.parse(value);


        String table=jsonObject.getString("table");
        String id=jsonObject.getObject("data",JSONObject.class).getString("id");;
        String line=jsonObject.getObject("data",JSONObject.class).toJSONString();

        String sql = "UPSERT INTO "+"dim_yk02_"+table+"(id, line) VALUES (?,?)";

        PreparedStatement ps =null;
        
        if(con!=null)
        {
            try {
                ps=con.prepareStatement(sql);
                ps.setString(1,id);
                ps.setString(2,line);


                ps.executeUpdate();
            }catch (Exception e)
            {
                e.printStackTrace();
            }finally {
                if(ps!=null)
                    ps.close();
            }
        }
    }

    @Override
    public void close() throws Exception {
        super.close();
        if(con!=null)
        {
            try {
                con.close();
            }catch (Exception e)
            {
                e.printStackTrace();
            }
        }
    }
}
